package com.lintf.distributed.transaction.order.task;

/**
 * @author lin tingfeng
 * @ system Mac
 * @ data 2025/6/12 17:49
 * @ package_name com.lintf.distributed.transaction.order.task
 * @ project_name distributed-transaction-demo
 * @ desc
 */

import com.lintf.distributed.transaction.order.entity.Message;
import com.lintf.distributed.transaction.order.service.MessageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class MessageRetryTask {

    private final MessageService messageService;
    private final KafkaTemplate<String, String> kafkaTemplate;

    // 每30秒执行一次重试
    @Scheduled(fixedRate = 30000)
    public void retryFailedMessages() {
        List<Message> pendingMessages = messageService.findMessagesByStatus(0); // 查找未发送的消息

        log.info("开始重试未发送的消息，数量: {}", pendingMessages.size());

        for (Message message : pendingMessages) {
            // 重试超过5次的消息标记为失败，需要人工介入
            if (message.getRetryCount() >= 5) {
                log.warn("消息重试超过5次，标记为失败: {}", message.getMessageId());
                message.setStatus(2); // 失败状态
                messageService.updateById(message);
                continue;
            }

            // 尝试重发消息
            trySendKafkaMessage(message);
        }
    }

    private void trySendKafkaMessage(Message message) {
        try {
            // 发送消息到Kafka
            kafkaTemplate.send(message.getQueueName(), message.getMessageId(), message.getContent())
                    .thenAccept(result -> {
                        log.info("消息发送成功: topic={}, key={}",
                                result.getRecordMetadata().topic(),
                                message.getMessageId());
                        message.setStatus(1); // 标记为已发送
                        messageService.updateById(message);
                    })
                    .exceptionally(ex -> {
                        log.error("消息发送失败: {}", message.getMessageId(), ex);
                        messageService.incrementRetryCount(message);
                        return null;
                    });

        } catch (Exception e) {
            log.error("消息发送异常: {}", message.getMessageId(), e);
            messageService.incrementRetryCount(message);
        }
    }
}
